package com.liuzhilong.infrastructure.framework.netty.api.standard.server;

import com.liuzhilong.infrastructure.framework.netty.api.standard.codec.CommandDecoder;
import com.liuzhilong.infrastructure.framework.netty.api.standard.codec.CommandEncoder;
import com.liuzhilong.infrastructure.framework.netty.api.standard.command.CommandType;
import com.liuzhilong.infrastructure.framework.netty.api.standard.handler.ServerHandler;
import com.liuzhilong.infrastructure.framework.netty.api.standard.processer.Processor;
import com.liuzhilong.infrastructure.framework.netty.api.standard.server.config.ServerConfig;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import com.liuzhilong.infrastructure.framework.netty.api.standard.command.Command;
/**
 * 一个基于{@link Command} 通信格式的服务端。
 * <p>
 *     使用{@link Processor} 作为处理器，增强扩展性和封装性。使用{@link Server#register(CommandType, Processor)} 注册一个处理器。
 *
 *     需要注意的是，处理器里面需要使用线程池进行处理自己的逻辑
 * </p>
 * 调用方式如下所示
 * <pre>
 *      ServerConfig config = new ServerConfig()
 *                 .setListenPort(19011)
 *                 .setSoKeepalive(true)
 *                 .setWorkerThread(4);
 *
 *      Server.create(config)
 *                 .register(CommandType.ECHO,new EchoService())
 *                 .start();
 * </pre>
 * @author longandai@163.com
 */
@Slf4j
public class Server {
    /**
     *  encoder
     */
    private final CommandEncoder encoder = new CommandEncoder();
    /**
     *  server bootstrap
     */
    private final ServerBootstrap serverBootstrap = new ServerBootstrap();

    /**
     * boss group
     */
    private final NioEventLoopGroup bossGroup;

    /**
     *  worker group
     */
    private final NioEventLoopGroup workGroup;
    /**
     * 服务配置
     */
    private ServerHandler serverHandler = new ServerHandler();
    /**
     *  配置信息
     */
    private  final ServerConfig serverConfig ;


    /**
     * started flag
     */
    private final AtomicBoolean isStarted = new AtomicBoolean(false);



    public Server(ServerConfig config){
        this.serverConfig =config;
        //设置Group 并设置Thread工厂

        this.bossGroup = new NioEventLoopGroup(1, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettyServerBossThread_%d", this.threadIndex.incrementAndGet()));
            }
        });

        this.workGroup = new NioEventLoopGroup(serverConfig.getWorkerThread(), new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettyServerWorkerThread_%d", this.threadIndex.incrementAndGet()));
            }
        });
    }


    //注册一个processor
    public Server register(CommandType commandType, Processor processor){
        serverHandler.getProcessorHolder().register(commandType,processor);
        return this;
    }

    /**
     * 停止
     */
    public void stop(){
        if(isStarted.compareAndSet(true, false)){
            try {
                if(bossGroup != null){
                    this.bossGroup.shutdownGracefully();
                }
                if(workGroup != null){
                    this.workGroup.shutdownGracefully();
                }
            } catch (Exception ex) {
                log.error("netty server close exception", ex);
            }
            log.info("netty server closed");
        }
    }
    /**
     * 启动
     */
    public void start() throws InterruptedException {
        if(this.isStarted.get()){
            return;
        }

        this.serverBootstrap
                .group(this.bossGroup, this.workGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_BACKLOG, serverConfig.getSoBacklog())
                .childOption(ChannelOption.SO_KEEPALIVE, serverConfig.isSoKeepalive())
                .childOption(ChannelOption.TCP_NODELAY, serverConfig.isTcpNoDelay())
                .childOption(ChannelOption.SO_SNDBUF, serverConfig.getSendBufferSize())
                .childOption(ChannelOption.SO_RCVBUF, serverConfig.getReceiveBufferSize())
                .childHandler(new ChannelInitializer<NioSocketChannel>() {

                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        initNettyChannel(ch);
                    }
                });

        ChannelFuture future;
        try {
            future = serverBootstrap.bind(serverConfig.getListenPort()).sync();
        } catch (Exception e) {
            log.error("NettyRemotingServer bind fail {}, exit",e.getMessage(), e);
            throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()));
        }
        if (future.isSuccess()) {
            log.info("NettyRemotingServer bind success at port : {}", serverConfig.getListenPort());
        } else if (future.cause() != null) {
            throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()), future.cause());
        } else {
            throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()));
        }
        //
        isStarted.compareAndSet(false, true);
    }




    /**
     *  init netty channel
     * @param ch socket channel
     * @throws Exception
     */
    private void initNettyChannel(NioSocketChannel ch) throws Exception{
        ChannelPipeline pipeline = ch.pipeline();
        // 编码可以复用
        pipeline.addLast("encoder", encoder);
        // 解码是每个channel一个 不能复用 里面有状态
        pipeline.addLast("decoder", new CommandDecoder());
        // 处理器是没有状态的 可以复用
        pipeline.addLast("handler", serverHandler);
    }
    public static final Server create(ServerConfig serverConfig){
        return new Server(serverConfig);
    }




}
